package com.sunday.common.mq.rocket.brave.study.base.e1_Normal_Message;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * https://rocketmq.apache.org/zh/docs/featureBehavior/01normalmessage
 * <p>
 * 普通消息
 * 普通消息为 Apache RocketMQ 中最基础的消息，区别于有特性的顺序消息、定时/延时消息和事务消息。
 * 本文为您介绍普通消息的应用场景、功能原理、使用方法和使用建议。
 * <p>
 * 应用场景
 * 普通消息一般应用于微服务解耦、事件驱动、数据集成等场景，
 * 这些场景大多数要求数据传输通道具有可靠传输的能力，且对消息的处理时机、处理顺序没有特别要求。
 * <p>
 * 典型场景一：微服务异步解耦
 * 在线的电商交易场景为例，上游订单系统将用户下单支付这一业务事件封装成独立的普通消息并发送至Apache RocketMQ服务端，
 * 下游按需从服务端订阅消息并按照本地消费逻辑处理下游任务。每个消息之间都是相互独立的，且不需要产生关联。
 * <p>
 * 典型场景二：数据集成传输
 * 离线的日志收集场景为例，通过埋点组件收集前端应用的相关操作日志，并转发到 Apache RocketMQ 。每条消息都是一段日志数据，
 * Apache RocketMQ 不做任何处理，只需要将日志数据可靠投递到下游的存储系统和分析系统即可，后续功能由后端应用完成。
 * <p>
 * 普通消息生命周期
 * 初始化：消息被生产者构建并完成初始化，待发送到服务端的状态。
 * 待消费：消息被发送到服务端，对消费者可见，等待消费者消费的状态。
 * 消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。 此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，Apache RocketMQ会对消息进行重试处理。具体信息，请参见消费重试。
 * 消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已经被处理（包括消费成功和失败）。 Apache RocketMQ默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。
 * 消息删除：Apache RocketMQ按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。更多信息，请参见消息存储和清理机制。
 * <p>
 * 使用限制
 * 普通消息仅支持使用MessageType为Normal主题，即普通消息只能发送至类型为普通消息的主题中，发送的消息的类型必须和主题的类型一致。
 * <p>
 * 2.3 消息的通信方式和流程
 * 在RocketMQ消息队列中支持通信的方式主要有同步(sync)、异步(async)、单向(oneway) 三种。其中“单向”通信模式相对简单，一般用在发送心跳包场景下，无需关注其Response。这里，主要介绍RocketMQ的异步通信流程。
 */
@Slf4j
public class NormalMessage {

    /**
     * 1.2 消息发送
     * 1、Producer端发送同步消息
     * 2、发送异步消息
     * 3、单向发送消息
     */

    /**
     * 1、Producer端发送同步消息
     * 这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。
     */
    @Test
    public void syncProducer() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("sync_producer_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 启动Producer实例
        producer.start();
        new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    // 创建消息，并指定Topic，Tag和消息体
                    Message msg = new Message(
                            "TopicTest" /* Topic */,
                            "TagA" /* Tag */,
                            ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                    );

                    // 发送消息到一个Broker
                    SendResult sendResult = producer.send(msg);
                    // 通过sendResult返回消息是否成功送达
                    log.info("{}", sendResult);
                    TimeUnit.SECONDS.sleep(5);
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }).start();

        // 如果不再发送消息，关闭Producer实例。
//        producer.shutdown();
    }

    /**
     * 2、发送异步消息
     * 异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。
     */
    @Test
    public void asyncProducer() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("async_producer_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 启动Producer实例
        producer.start();
        // 在异步模式下，在声明发送失败之前，内部执行的最大重试次数。
        // 这可能会导致消息重复，需要应用程序开发人员解决。
        producer.setRetryTimesWhenSendAsyncFailed(0);

        int messageCount = 10;
        // 根据消息数量实例化倒计时计算器
        final CountDownLatch2 countDownLatch = new CountDownLatch2(messageCount);

        for (int i = 0; i < messageCount; i++) {
            final int index = i;
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message(
                    "TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            // SendCallback接收异步返回结果的回调
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    countDownLatch.countDown();
                    log.info("{} OK {}", index, sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable e) {
                    countDownLatch.countDown();
                    log.info("{} Exception {}", index, e);
                    e.printStackTrace();
                }
            });
        }

        // 等待5s
        countDownLatch.await(5, TimeUnit.SECONDS);
        // 如果不再发送消息，关闭Producer实例。
//        producer.shutdown();
    }

    /**
     * 3、单向发送消息
     * 这种方式主要用在不特别关心发送结果的场景，例如日志发送。
     */
    @Test
    public void onewayProducer() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("oneway_producer_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 启动Producer实例
        producer.start();
        new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    // 创建消息，并指定Topic，Tag和消息体
                    Message msg = new Message(
                            "TopicTest" /* Topic */,
                            "TagA" /* Tag */,
                            ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                    );
                    // 发送单向消息，没有任何返回结果
                    producer.sendOneway(msg);
                    TimeUnit.SECONDS.sleep(5);
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }).start();

        // 如果不再发送消息，关闭Producer实例。
//        producer.shutdown();
    }

    /**
     * 1.3 消费消息
     */
    @AfterEach
    public void consumer() throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group_name");
        /**
         * MessageModel 定义了两个常量：BROADCASTING 和 CLUSTERING，并分别对应 "广播" 和 "集群" 两种消息模式。
         * BROADCASTING 表示广播模式，也称为订阅-发布/广播模式。在这种模式下，消息发送者将消息发布到一个主题（或通道），所有已经订阅该主题的接收者都会接收到该消息。每个消息可以被多个接收者消费。
         * CLUSTERING 表示集群模式。在集群消费模式中，多个消费者共同组成一个集群，并通过某种机制来协调消息的消费。消息中间件将消息分发给具有相同消费组标识的消费者实例，以确保每个消息只被一个消费者处理。这种模式可以提高系统的吞吐量、实现负载均衡和故障容错。
         */
        consumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * https://github.com/apache/rocketmq/blob/master/docs/cn/best_practice.md
         *
         * 2.2 消费速度慢的处理方式
         * 1 提高消费并行度
         * 绝大部分消息消费行为都属于 IO 密集型，即可能是操作数据库，或者调用 RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，通过增加消费并行度，
         * 可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。所以，应用必须要设置合理的并行度。 如下有几种修改消费并行度的方法：
         *
         * 同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。
         * 可以通过加机器，或者在已有机器启动多个进程的方式。
         * 提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax实现。
         */
//        consumer.setConsumeThreadMin(1); // default 20
//        consumer.setConsumeThreadMax(1); // default 20

        /**
         * 2 批量方式消费
         * 某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时 1 s，一次处理 10 个订单可能也只耗时 2 s，
         * 这样即可大幅度提高消费的吞吐量，
         * 通过设置 consumer的 consumeMessageBatchMaxSize 返个参数，默认是 1，即一次只消费一条消息，例如设置为 N，那么每次消费的消息数小于等于 N。
         *
         * 也就是 consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) 中的 msgs 为什么是一个list
         */
//        consumer.setConsumeMessageBatchMaxSize(10); // default 1

        // 设置NameServer的地址
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("TopicTest", "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                log.info("Receive New Messages: {}", msgs.size());
//                try {
//                    TimeUnit.SECONDS.sleep(10);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        log.info("Consumer Started.");

        TimeUnit.SECONDS.sleep(60);
    }

}
